package com.stormragetech.demo.hbase.spring.dao.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * description: //TODO
 *
 * @author: Xue Bin
 * @version: 1.0.0-SNAPSHOT
 */
@Repository
public abstract class HBaseBaseDao {

    private final static String ZK_PORT_K = "hbase.zookeeper.property.clientPort";
    private final static String ZK_PORT_V = "2181";

    private final static String ZK_HOST_K = "hbase.zookeeper.quorum";
    private final static String ZK_HOST_v = "10.12.99.1";

//    private final static String HBASE_MASTER_K = "hbase.master";
//    private final static String HBASE_MASTER_V = "192.168.1.21:60000";

    private final static int threadCount = 5;

    private static Configuration conf;
    private static Connection conn;

    static {
        conf = HBaseConfiguration.create();
        conf.set(ZK_PORT_K, ZK_PORT_V);
        conf.set(ZK_HOST_K, ZK_HOST_v);
//        conf.set("hbase.master", "192.168.1.21:60000");
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        try {
            conn = ConnectionFactory.createConnection(conf, executor);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void createTable(String table, String family) throws IOException {
        Admin admin = conn.getAdmin();
        System.out.println((null == admin) + " || " + admin.listTableNames());
        if (admin.tableExists(TableName.valueOf(table))) {
            if (admin.isTableEnabled(TableName.valueOf(table))) {
                admin.disableTable(TableName.valueOf(table));
            }
            admin.deleteTable(TableName.valueOf(table));
        }
        HTableDescriptor descriptor = new HTableDescriptor(table);
        descriptor.addFamily(new HColumnDescriptor(family));
        admin.createTable(descriptor);
        admin.close();
    }

    public void delTable(String table) throws IOException {
        Admin admin = conn.getAdmin();
        admin.disableTable(TableName.valueOf(table));
        admin.deleteTable(TableName.valueOf(table));
        admin.close();
    }

    public void put(String id, String column, String value, String family, String table) throws IOException {

        Table t = conn.getTable(TableName.valueOf(table));
        Put put = new Put(Bytes.toBytes(id));
        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
        t.put(put);
        t.close();
//        conn.close();
    }

    public Result get(String id, String column, String family, String table) throws IOException {
        Table t = conn.getTable(TableName.valueOf(table));
        Get get = new Get(Bytes.toBytes(id));
        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
        get.setMaxVersions(1);
        return t.get(get);
    }

}
